package com.mall.util.distributed.raft.heartbeat;

import com.mall.util.distributed.raft.node.Node;
import com.mall.util.distributed.raft.node.NodeRole;
import com.mall.util.distributed.raft.rpc.RaftClient;
import com.mall.util.distributed.raft.rpc.Request;
import com.mall.util.distributed.raft.rpc.message.AppendEntries;
import com.mall.util.distributed.raft.rpc.message.ResponseAppendEntries;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@RequiredArgsConstructor
public class HeaderBeatTask implements Runnable {
    private final ExecutorService executorService;

    @Autowired
    public RaftClient raftClient;

    @Override
    public void run() {
        var node = Node.getINSTANCE();
        var preHeartBeatTime = node.getPreHeartBeatTime();
        var heartBeatTick = node.getHeartBeatTick();

        /* 只有Leader结点才有心跳任务 */
        if(!node.isLeader()) return;

        var current = System.currentTimeMillis();
        if(current - preHeartBeatTime < heartBeatTick) return;

        node.setPreHeartBeatTime(current);

        log.info("======================= NextIndex =======================");
        for(String peer : node.getExcludeSelfPeerAddress()) {
            log.info("Peer {} nextIndex {}", peer, node.getNextIndexs().get(peer));
        }

        /* 心跳只关心term和leader Id */
        for(String peer : node.getExcludeSelfPeerAddress()) {
            var appendEntries = new AppendEntries();
            appendEntries.setLeaderId(node.getSelfAddress());
            appendEntries.setTerm(node.getCurrentTerm().get());
            appendEntries.setEntries(null);

            var request = new Request<AppendEntries>(
                    Request.A_ENTRIES,
                    appendEntries,
                    peer);

            /* 异步处理心跳信息 */
            executorService.execute(() -> {
                var response = raftClient.send(request);
                var responseAppendEntries = (ResponseAppendEntries) response.getResult();

                synchronized (this) {
                    long term = responseAppendEntries.getTerm();
                    if (term > node.getCurrentTerm().get()) {
                        log.info("Current node change to follower role, he's term : {}, my term : {}", term, node.getCurrentTerm().get());
                        node.setNodeRole(NodeRole.FOLLOWER);
                        node.setCurrentTerm(new AtomicLong(term));
                        node.resetMark();
                    }
                }
            });
        }
    }
}
